package com.bwie;

/**
 * @version 1.0
 * @Author 佩奇🍂
 * @Date 2024/9/23 16:24
 * @注释
 */
public class Main {
    public static void main(String[] args) {
        System.out.println("Hello world!");
    }
    //登录
    //登出

    @Autowired
    KafkaTemplate kafkaTemplate;

    @Autowired
    KafkaProducer kafkaProducer;

    private Long orderId = 0L;
    @GetMapping(value = {
            "sendMessageObject/{userName}/{price}/{partition}",
            "sendMessageObject/{userName}/{price}"
    })
    public Order sendMessageObject(
            @PathVariable String userName,
                        @PathVariable Integer price,
                        @PathVariable(required = false) Integer partition) {
        Order order = new Order();
        order.setId(orderId++);
        order.setUserName(userName);
        order.setName("订单:" + orderId);
        order.setPrice(price);
        System.out.println("待处理的订单已纳入处理队列（kafka），id：" + orderId);


        /* @Author 佩奇♥
         * @Description //TODO
         * @Date         * topic 主题
         * key
         * value
         * 没有指定分片：策略
         */
        if (partition == null) {
            kafkaTemplate.send("my-2109A-topic", userName, order);//使用send方法发送消息，需要传入topic名称
        }
        kafkaTemplate.send("my-2109A-topic", partition, userName, order);
        /*
        kafkaTemplate.send("my-2109A-topic", 0,userName,order);
        kafkaTemplate.send("my-2012A-topic", 1,userName,order);
        */

        return order;
    }

    @GetMapping("sendMessageObject1/{userName}/{price}")
    public Order sendMessageObject1(@PathVariable String userName,
                                                                       @PathVariable Integer price) {

        Order order = new Order();
        order.setId(orderId++);
        order.setUserName(userName);
        order.setName("订单:" + orderId);
        order.setPrice(price);
        System.out.println("待处理的订单已纳入处理队列（kafka），id：" + orderId);

        ProducerRecord<String, String> producerRecord =
                new ProducerRecord("my-2109A-topic", userName, JSON.toJSONString(order));
        kafkaProducer.send(producerRecord);
        /*
        ProducerRecord<String, String> producerRecord =
                new ProducerRecord("my-2109A-topic", 0,
                        userName, JSON.toJSONString(order));
        kafkaProducer.send(producerRecord);

        ProducerRecord<String, String> producerRecord1 =
                new ProducerRecord("my-2109A-topic", 1,
                        userName, JSON.toJSONString(order));
        kafkaProducer.send(producerRecord1);*/

        return order;

    }

@Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Autowired
    @Qualifier("kafkaTranProducer")
    private KafkaProducer<String, String> kafkaTranProducer;

    @GetMapping("sendMessageObject/{id}/{name}/{age}")
    public void sendMessageObject(@PathVariable Long id,
                                                              @PathVariable String name,
                            @PathVariable Integer price) {

        Order order = new Order();
        order.setId(id);
        order.setName(name);
        order.setPrice(price);
        System.out.println("待发送短信的订单已纳入处理队列（kafka），id：" + id);

        ProducerRecord producerRecord =
                new ProducerRecord("my-2109A-topic", "user_123", JSON.toJSONString(order));
        kafkaTranProducer.initTransactions();
        try {
            kafkaTranProducer.beginTransaction();

            for (int i = 0; i < 10; i++) {
                kafkaTranProducer.send(producerRecord);
            }
            int i = 1 / 0;
            kafkaTranProducer.commitTransaction();
        } catch (Exception e) {
            kafkaTranProducer.abortTransaction();
        }
    }

//静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    private static final AtomicInteger onlineNum = new AtomicInteger(0);
    //concurrent包的线程安全Map，用来存放每个客户端对应的Session对象。
    private static final Set<Session> connections = new CopyOnWriteArraySet<>();

    public static final Map<String,Session> userConnections = new ConcurrentHashMap<>();
    private String loginName;
    private Session session;

    //处理四种事件
    //OnOpen  客户端发起连接 ，服务器端OnOpen响应
    //OnClose 客户端断开连接 ，服务器端OnCLose响应
    //OnMessage 客户端发送消息，服务器端OnMessage响应
    //OnError  客户端发生错误，服务器端OnError响应

    //关闭连接close()
    //发送消息send()

    // 处理连接建立
    @OnOpen
    public void onOpen(Session session,@PathParam("loginName")String loginName ) {
        // 获取登录名
        this.loginName = loginName;
        this.session = session;
        connections.add(session);
        userConnections.put(loginName, session);
        onlineNum.incrementAndGet();//i++
        String msg = String.format("欢迎 %s(%s)加入，当前在线人数 %d", loginName, session.toString(), onlineNum.get());
        System.out.println(msg);

        broadcast(loginName + " 加入聊天室");
    }

    // 处理连接关闭
    @OnClose
    public void onClose(Session session,@PathParam("loginName")String loginName) {
        connections.remove(session);
        onlineNum.decrementAndGet();
        userConnections.remove(loginName);
        String msg = String.format("%s(%s) 退出，当前在线人数 %d", loginName, session.toString(), onlineNum.get());
        System.out.println(msg);

        broadcast(loginName + " 退出聊天室");
    }

    /* @Author 佩奇♥
     * @Description //TODO
     * @Date     * 接收到消息的监听方法
          * client write    ------>   server  read
               * server write    ------>   client  read
                    * @param message  自定义协议
                         *         {//群聊
                              *             "to":'all',
                                   *             "content":'hello'
                                        *         }
                                             *         {//私聊
                                                  *             "to":'zhangsan',
                                                       *             "content":'hello'
                                                            *         }
                                                                 */
    @OnMessage
    public void onMessage(Session session,String message,@PathParam("loginName")String loginName) {
        String msg = String.format("发送者%s(%s)发送消息： %s", loginName, session.toString(), message);
        System.out.println(msg);
        //解析消息类型

        JSONObject jsonObject = JSON.parseObject(message);
        String to = (String) jsonObject.get("to");
        String content = (String) jsonObject.get("content");
        if (to.equalsIgnoreCase("all")) {
            broadcast(loginName + ":" + HTMLFilter.filter(content));
        } else {
            Session sessionTo = userConnections.get(to);
            sessionTo.getAsyncRemote().sendText(loginName + ":" + HTMLFilter.filter(content));
        }
    }

22:15 2024/9/23
         * @param null
     * @return @return null
     **/
    }// 22:14 2024/9/23
             * @param userName
 * @param price
 * @param partition
         * @return @return Order
         **/
    }
    )
    })
}